/**
 * Copyright (c) 2013-2022 Nikita Koksharov
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson.pubsub;

import org.redisson.RedissonLockEntry;

import java.util.concurrent.CompletableFuture;

/**
 * LockPubSub 类继承自 PublishSubscribe 类，专门用于处理 Redisson 锁的发布与订阅功能。
 * RedissonLockEntry 是锁信息的实体类。
 */
public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {


    /**
     * 普通锁解锁标志
     */
    public static final Long UNLOCK_MESSAGE = 0L;

    /**
     * 读锁解锁标志
     */
    public static final Long READ_UNLOCK_MESSAGE = 1L;





    public LockPubSub(PublishSubscribeService service) {
        super(service);
    }
    
    @Override
    protected RedissonLockEntry createEntry(CompletableFuture<RedissonLockEntry> newPromise) {
        return new RedissonLockEntry(newPromise);
    }



    /**
     * 当收到消息时处理Redisson锁的逻辑。
     * @param value RedissonLockEntry对象，代表一个锁的实例，包含锁的状态和监听器等信息。
     * @param message 收到的消息内容，用于判断是普通解锁还是读锁解锁。
     */
    @Override
    protected void onMessage(RedissonLockEntry value, Long message) {
        // 检查是否接收到普通解锁消息
        if (message.equals(UNLOCK_MESSAGE)) {
            // 从监听器队列取监听线程执行监听回调
            Runnable runnableToExecute = value.getListeners().poll();
            if (runnableToExecute != null) {
                runnableToExecute.run(); // 执行监听回调
            }
            value.getLatch().release(); // 释放信号量
        }

        // 检查是否接收到读锁解锁消息
        else if (message.equals(READ_UNLOCK_MESSAGE)) {
            while (true) { // 循环处理所有等待的读锁监听器
                Runnable runnableToExecute = value.getListeners().poll(); // 尝试从监听器队列中取出一个监听器
                if (runnableToExecute == null) {
                    break; // 如果没有监听器了，则退出循环
                }
                runnableToExecute.run(); // 执行监听器
            }

            value.getLatch().release(value.getLatch().getQueueLength()); // 释放锁，释放的数量为当前监听器队列的长度
        }
    }

}
